package com.kx.smsplatform.deadletter.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.kx.smsplatform.client.CacheClient;
import com.kx.smsplatform.common.constants.CacheConstants;
import com.kx.smsplatform.common.constants.RabbitMQConstants;
import com.kx.smsplatform.common.model.StandardReport;
import com.kx.smsplatform.deadletter.config.DelayedConfig;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;
import java.util.Map;

/**
 * @Author Sun
 * @Date 2021/7/8 16:46
 */
@Slf4j
@Component
public class ListenerPushReport {

    @Autowired
    private CacheClient cacheClient;

    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private ObjectMapper objectMapper;

    private Integer[] sendTTL = {0, 10000, 30000, 60000, 300000};


    //监听push_sms_report_topic队列，给用户推送状态报告
    @RabbitListener(queues = RabbitMQConstants.TOPIC_PUSH_SMS_REPORT)
    public void consume(StandardReport report, Channel channel, Message message) throws IOException {
        long clientID = report.getClientID();
        log.info("【接口模块 推送状态报告】 给用户user={},推送状态报告开始执行", clientID);
        //1. 查询客户信息
        Map map = cacheClient.hGetAll(CacheConstants.CACHE_PREFIX_CLIENT + clientID);

        //2. 根据客户信息中的isreturnstatus判断客户是否需要推送状态报告
        Integer isreturnstatus = (Integer) map.get("isreturnstatus");
        if (isreturnstatus !=1) {//说明不需要推送状态报告
            // 手动ack，并且return。
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            return;
        }

        //3. 如果需要推送，获取到客户的receivestatusurl
        String url = (String) map.get("receivestatusurl");
        report.setReceiveStatusUrl(url);

        //4. 给receivestatusurl发送一个HTTP请求，只要客户接收到请求，并响应SUCCESS，证明客户成功接收
        log.info("【接口模块 推送状态报告】 当前第 {} 次，给客户推送状态报告",report.getSendCount());
        //4.1创建响应头,封装响应参数
        HttpHeaders httpHeaders = new HttpHeaders();
        httpHeaders.setContentType(MediaType.APPLICATION_JSON_UTF8);
        HttpEntity httpEntity = new HttpEntity(objectMapper.writeValueAsString(report),httpHeaders);
        //4.2 发送请求
        String result = null;
        try {
            result = restTemplate.postForObject(url, httpEntity, String.class);
        } catch (RestClientException e) {
            sendDelayedMessage(report);
            //6. 手动Ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            return ;
        }
        //4.3 判断返回的是否是SUCCESS
        if(StringUtils.isEmpty(result) || !result.equals("SUCCESS")){
            sendDelayedMessage(report);
        }

        //6. 手动Ack
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

        //4.3 判断返回的是否是SUCCESS
        if(StringUtils.isEmpty(result) || !result.equals("SUCCESS")){
            sendDelayedMessage(report);
        }

    }



    // 第二次,三次,四次,五次,六次的推送,监听delayed-queue
    @RabbitListener(queues = DelayedConfig.DELAYED_QUEUE)
    public void delayedConsume(StandardReport report, Channel channel, Message message) throws Exception {
        //至少是第二次推送消息 5
        log.info("【接口模块-推送状态报告】 当前第 {} 次，给客户推送状态报告",report.getSendCount());
        //1. 封装参数和请求头
        HttpHeaders header = new HttpHeaders();
        header.setContentType(MediaType.APPLICATION_JSON_UTF8);
        HttpEntity httpEntity = new HttpEntity(objectMapper.writeValueAsString(report),header);

        //2. 通过restTemplate发送请求
        String result = null;
        try {
            result = restTemplate.postForObject(report.getReceiveStatusUrl(), httpEntity, String.class);
        } catch (RestClientException e) {
            // 推送状态报告失败
            //4. 如果已经完成了第6次推送,不需要再次发送消息到延迟队列,      再次发送消息给延迟队列
            if(report.getSendCount() < 5) {  // 0  1  2  3  4  5
                sendDelayedMessage(report);
            }
            //6. 手动Ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            return ;
        }

        //3. 如果走了catch或者返回结果不是SUCCESS
        if(StringUtils.isEmpty(result) || !result.equals("SUCCESS")){
            // 推送状态报告失败
            // 如果已经完成了第6次推送,不需要再次发送消息到延迟队列,再次发送消息给延迟队列
            if(report.getSendCount() < 5)
                sendDelayedMessage(report);
        }

        //4. 手动ack
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }



    // 推送失败时,发消息给延迟队列
    private void sendDelayedMessage(StandardReport report) {
        // 发送失败
        report.setSendCount(report.getSendCount() + 1);
        //5. 如果客户没收接收到，需要尝试再次推送（隔一段时间一次,基于RabbitMQ实现）
        rabbitTemplate.convertAndSend(DelayedConfig.DELAYED_EXCHANGE, "", report, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setDelay(sendTTL[report.getSendCount()]);
                return message;
            }
        });
    }
}
